package kq

import (
	"context"
	"encoding/json"
	"fmt"
	"strconv"
	"strings"
	"time"

	"mall/common/globalkey"
	"mall/common/modelx"
	"mall/common/mqueuex"
	"mall/common/resultx"
	"mall/common/subscribex"
	"mall/common/utils/pointy"
	"mall/service/backend/rmq/internal/svc"
	"mall/service/promotion/rpc/promotionclient"
	"mall/service/setting/rpc/settingclient"
	"mall/service/user/rpc/userclient"

	// "github.com/sirupsen/logrus"
	"github.com/dtm-labs/client/dtmcli"
	"github.com/dtm-labs/client/dtmcli/dtmimp"
	"github.com/dtm-labs/client/dtmgrpc"
	"github.com/silenceper/wechat/v2/miniprogram/subscribe"
	"github.com/zeromicro/go-zero/core/logx"
	"github.com/zeromicro/go-zero/core/service"
	"google.golang.org/grpc/codes"
)

/**
 * Listening to the payment flow status change notification message queue
 */
type DistributePointsMq struct {
	ctx    context.Context
	svcCtx *svc.ServiceContext
}

func NewDistributePointsMq(ctx context.Context, svcCtx *svc.ServiceContext) *DistributePointsMq {
	return &DistributePointsMq{
		ctx:    ctx,
		svcCtx: svcCtx,
	}
}

func (l *DistributePointsMq) Consume(_, val string) error {
	// 解析数据
	var message mqueuex.DistributePointsMessage
	if err := json.Unmarshal([]byte(val), &message); err != nil {
		logx.Errorw("DistributePointsMq.Consume json.Unmarshal err", logx.Field("detail", err.Error()))
		return err
	}

	// logrus.Info(fmt.Sprintf("DistributePointsMq message: %+v", message))

	// 执行业务..
	if err := l.execService(&message); err != nil {
		logx.Errorw("DistributePointsMq.execService err", logx.Field("detail", err.Error()),
			logx.Field("val", val), logx.Field("message", fmt.Sprintf("%+v", message)))
		return err
	}

	return nil
}

// Verify and update relevant flow data
func (l *DistributePointsMq) execService(message *mqueuex.DistributePointsMessage) error {
	if message.UserId != nil && *message.UserId > 0 {
		if message.TaskId != nil && *message.TaskId > 0 &&
			message.SourceType != nil && *message.SourceType == globalkey.UserAssetSourceTypeReward {
			var sid uint64
			if message.SourceId != nil && *message.SourceId > 0 {
				sid = *message.SourceId
			}
			//查询任务信息
			task, err := l.svcCtx.SettingRpc.GetIntegralTaskById(l.ctx, &settingclient.IntegralTaskDetailReq{
				Id: message.TaskId,
			})
			if err != nil {
				return fmt.Errorf("DistributePointsMq.execService GetIntegralTaskById err: %+v", err)
			}

			if task.Points == nil || *task.Points == 0 {
				return fmt.Errorf("DistributePointsMq.execService Points is empty")
			}

			if task.RuleName == nil || len(strings.TrimSpace(*task.RuleName)) == 0 {
				return fmt.Errorf("DistributePointsMq.execService RuleName is empty")
			}

			amount := strconv.FormatUint(*task.Points, 10)

			err = l.distributePointsDtmSaga(message.UserId, &userclient.UpdateUserAssetReq{
				UserId:       message.UserId,
				AssetType:    pointy.GetPointer(globalkey.UserAssetTypePoint),          // 积分
				ChangeType:   pointy.GetPointer(globalkey.UserAssetChangeTypeIncrease), // 增加
				ChangeAmount: pointy.GetPointer(amount),
				SourceType:   pointy.GetPointer(globalkey.UserAssetSourceTypeTask), // 任务
				SourceId:     pointy.GetPointer(sid),
				SourceDesc:   task.RuleName,
				TaskId:       task.Id,
				TaskName:     task.RuleName,
				TaskSlug:     task.RuleSlug,
			})
			if err != nil {
				return err
			}

			return nil
		}

		if message.Points != nil && len(strings.TrimSpace(*message.Points)) > 0 &&
			message.SourceType != nil && *message.SourceType == globalkey.UserAssetSourceTypeReward {
			err := l.distributePointsDtmSaga(message.UserId, &userclient.UpdateUserAssetReq{
				UserId:       message.UserId,
				AssetType:    pointy.GetPointer(globalkey.UserAssetTypePoint),          // 积分
				ChangeType:   pointy.GetPointer(globalkey.UserAssetChangeTypeIncrease), // 增加
				ChangeAmount: message.Points,
				SourceType:   pointy.GetPointer(globalkey.UserAssetSourceTypeReward), // 奖励
				SourceId:     message.SourceId,
				SourceDesc:   pointy.GetPointer("消费积分"),
			})
			if err != nil {
				return err
			}

			return nil
		}

		if message.Coupons != nil && len(strings.TrimSpace(*message.Coupons)) > 0 &&
			message.SourceType != nil && *message.SourceType == globalkey.UserAssetSourceTypeCoupon {
			err := l.deductPointsForCouponDtmSaga(message)
			if err != nil {
				return err
			}

			return nil
		}

		if message.Product != nil && len(strings.TrimSpace(*message.Product)) > 0 &&
			message.SourceType != nil && *message.SourceType == globalkey.UserAssetSourceTypeProduct {
			err := l.deductPointsForProductDtmSaga(message)
			if err != nil {
				return err
			}

			return nil
		}
	}

	return fmt.Errorf("DistributePointsMq.execService UserId OR TaskId is empty")
}

func (l *DistributePointsMq) distributePointsDtmSaga(
	uid *uint64,
	userAsset *userclient.UpdateUserAssetReq,
) error {
	// 获取 UserRpc BuildTarget
	userRpcBusiServer, err := l.svcCtx.Config.UserRpc.BuildTarget()
	if err != nil {
		return err
	}

	// DefaultGrpcServer default url for grpc server.
	defaultGrpcServer := l.svcCtx.Config.DefaultGrpcServer

	// 创建一个gid
	gid := dtmgrpc.MustGenGid(defaultGrpcServer)

	// 创建一个saga协议的事务
	saga := dtmgrpc.NewSagaGrpc(defaultGrpcServer, gid).
		Add(
			userRpcBusiServer+"/user.User/UpdateUserAssetRaw",
			userRpcBusiServer+"/user.User/UpdateUserAssetRevert",
			userAsset,
		)

	// 事务提交
	err = saga.Submit()
	if err != nil {
		// logrus.Info(fmt.Sprintf("DistributePointsMq saga.Submit err: %+v", err))
		logx.Errorw("DistributePointsMq distributePointsDtmSaga saga.Submit error", logx.Field("detail", err.Error()))
	}

	dtmimp.FatalIfError(err)
	if err != nil {
		logx.Errorw("DistributePointsMq distributePointsDtmSaga error", logx.Field("detail", err.Error()))
		return err
	}

	//此后无论任何报错或参数错误，打印日志并返回nil，不可影响dtm执行
	//1、获取用户openID
	uids := []uint64{*uid}
	authList, err := l.svcCtx.UserRpc.BackendGetUserAuthList(l.ctx, &userclient.BackendUserAuthListReq{
		Ids: uids,
	})
	if err != nil {
		// logx.Errorw(fmt.Sprintf("DistributePointsMq BackendGetUserAuthList err: %+v", err))
		logx.Errorw("DistributePointsMq distributePointsDtmSaga BackendGetUserAuthList error", logx.Field("detail", err.Error()))
		return nil
	}

	var openId string
	if authList != nil && len(authList.List) > 0 {
		auth := authList.List[0]
		if auth != nil && auth.AuthKey != nil && len(strings.TrimSpace(*auth.AuthKey)) > 0 {
			openId = *auth.AuthKey
		}
	}

	// 完成积分发放 未通知，记录日志
	if len(strings.TrimSpace(openId)) == 0 {
		logx.Errorw(fmt.Sprintf("DistributePointsMq distributePointsDtmSaga UserId: %+v, openId is empty", *uid))
		return nil
	}

	//2、获取用户通知消息并发送
	messages := l.getMessages(openId, userAsset)
	if len(messages) > 0 {
		err = l.sendNotificationMessage(messages[0])
		if err != nil {
			// logx.Errorw(fmt.Sprintf("DistributePointsMq sendNotificationMessage err: %+v", err))
			logx.Errorw("DistributePointsMq distributePointsDtmSaga sendNotificationMessage error", logx.Field("detail", err.Error()))
		}
	}

	return nil
}

func (l *DistributePointsMq) deductPointsForCouponDtmSaga(
	message *mqueuex.DistributePointsMessage,
) error {
	if message.Coupons == nil || len(strings.TrimSpace(*message.Coupons)) == 0 {
		return nil
	}

	var coupons []*mqueuex.CouponInfo
	// 不再重试，走回滚
	if err := json.Unmarshal([]byte(strings.TrimSpace(*message.Coupons)), &coupons); err != nil {
		logx.Errorw("DistributePointsMq deductPointsForCouponDtmSaga json.Unmarshal err", logx.Field("detail", err.Error()))
		return resultx.StatusError(uint32(codes.Aborted), dtmcli.ResultFailure)
	}

	// 获取 UserRpc BuildTarget
	userRpcBusiServer, err := l.svcCtx.Config.UserRpc.BuildTarget()
	if err != nil {
		return err
	}

	promotionRpcBusiServer, err := l.svcCtx.Config.PromotionRpc.BuildTarget()
	if err != nil {
		return err
	}

	// DefaultGrpcServer default url for grpc server.
	defaultGrpcServer := l.svcCtx.Config.DefaultGrpcServer

	for _, v := range coupons {
		// 创建一个gid
		gid := dtmgrpc.MustGenGid(defaultGrpcServer)

		if v.Point == nil || len(*v.Point) == 0 {
			continue
		}

		userAsset := &userclient.UpdateUserAssetReq{
			UserId:       message.UserId,
			AssetType:    pointy.GetPointer(globalkey.UserAssetTypePoint),          // 积分
			ChangeType:   pointy.GetPointer(globalkey.UserAssetChangeTypeDecrease), // 减少
			ChangeAmount: v.Point,
			SourceType:   pointy.GetPointer(globalkey.UserAssetSourceTypeCoupon), // 兑换优惠券
			SourceId:     pointy.GetPointer(v.TemplateId),
			SourceDesc:   pointy.GetPointer("兑换优惠券"),
		}
		coupon := &promotionclient.CouponInfo{
			ActivityId: v.ActivityId,
			TemplateId: pointy.GetPointer(v.TemplateId),
			UserId:     pointy.GetPointer(v.UserId),
			StartedAt:  pointy.GetPointer(v.StartedAt),
			EndedAt:    pointy.GetPointer(v.EndedAt),
		}

		// 创建一个saga协议的事务
		saga := dtmgrpc.NewSagaGrpc(defaultGrpcServer, gid).
			Add(
				userRpcBusiServer+"/user.User/UpdateUserAssetRaw",
				userRpcBusiServer+"/user.User/UpdateUserAssetRevert",
				userAsset,
			).
			Add(
				promotionRpcBusiServer+"/promotion.Promotion/ExchangeCouponRaw",
				promotionRpcBusiServer+"/promotion.Promotion/ExchangeCouponRevert",
				coupon,
			)

		// 事务提交
		err = saga.Submit()
		if err != nil {
			// logrus.Info(fmt.Sprintf("DistributePointsMq saga.Submit err: %+v", err))
			logx.Errorw("DistributePointsMq distributePointsForCouponDtmSaga saga.Submit error", logx.Field("detail", err.Error()))
		}

		dtmimp.FatalIfError(err)
		if err != nil {
			logx.Errorw("DistributePointsMq distributePointsForCouponDtmSaga error", logx.Field("detail", err.Error()))
			return err
		}
	}

	return nil
}

func (l *DistributePointsMq) deductPointsForProductDtmSaga(
	message *mqueuex.DistributePointsMessage,
) error {
	if message.Product == nil || len(strings.TrimSpace(*message.Product)) == 0 {
		return nil
	}

	var product *mqueuex.ProductInfo
	// 不再重试，走回滚
	if err := json.Unmarshal([]byte(strings.TrimSpace(*message.Product)), &product); err != nil {
		logx.Errorw("DistributePointsMq deductPointsForProductDtmSaga json.Unmarshal err", logx.Field("detail", err.Error()))
		return resultx.StatusError(uint32(codes.Aborted), dtmcli.ResultFailure)
	}

	if product.Point == nil || len(*product.Point) == 0 {
		return nil
	}

	// 获取 UserRpc BuildTarget
	userRpcBusiServer, err := l.svcCtx.Config.UserRpc.BuildTarget()
	if err != nil {
		return err
	}

	// DefaultGrpcServer default url for grpc server.
	defaultGrpcServer := l.svcCtx.Config.DefaultGrpcServer

	gid := dtmgrpc.MustGenGid(defaultGrpcServer)

	userAsset := &userclient.UpdateUserAssetReq{
		UserId:       message.UserId,
		AssetType:    pointy.GetPointer(globalkey.UserAssetTypePoint),          // 积分
		ChangeType:   pointy.GetPointer(globalkey.UserAssetChangeTypeDecrease), // 减少
		ChangeAmount: product.Point,
		SourceType:   pointy.GetPointer(globalkey.UserAssetSourceTypeProduct), // 兑换商品
		SourceId:     pointy.GetPointer(product.SkuId),
		SourceDesc:   message.Product,
	}

	// 创建一个saga协议的事务
	saga := dtmgrpc.NewSagaGrpc(defaultGrpcServer, gid).
		Add(
			userRpcBusiServer+"/user.User/UpdateUserAssetRaw",
			userRpcBusiServer+"/user.User/UpdateUserAssetRevert",
			userAsset,
		)

	// 事务提交
	err = saga.Submit()
	if err != nil {
		// logrus.Info(fmt.Sprintf("DistributePointsMq saga.Submit err: %+v", err))
		logx.Errorw("DistributePointsMq deductPointsForProductDtmSaga saga.Submit error", logx.Field("detail", err.Error()))
	}

	dtmimp.FatalIfError(err)
	if err != nil {
		logx.Errorw("DistributePointsMq deductPointsForProductDtmSaga error", logx.Field("detail", err.Error()))
		return err
	}

	return nil
}

// 发送通知消息
func (l *DistributePointsMq) sendNotificationMessage(
	msg *subscribe.Message,
) error {
	if l.svcCtx.Config.Mode != service.ProMode {
		msg.MiniprogramState = "developer"
	} else {
		msg.MiniprogramState = "formal"
	}

	err := l.svcCtx.MiniProgram.GetSubscribe().Send(msg)
	if err != nil {
		// logrus.Info(fmt.Sprintf("DistributePointsMq send wechat mini subscription message failed, err: %+v, msg: %+v", err, msg))
		// err: Send Error , errcode=47003 , errmsg=argument invalid! data.thing1.value is empty rid: 64b8e27d-4c575776-556e0d30, msg: &{ToUser:o-kUG5pMqzsclBbFlUdZNJ2OW4_o TemplateID:4IledrS3WBJMOfaS5NeWNZ9Pq1yibDuvY0woYHhyUsc Page:pages/publicPage/message-page Data:map[date3:0xc000995de0 phrase5:0xc000995d80 thing1:0xc000995da0 thing2:0xc000995dc0 thing4:0xc000995e00] MiniprogramState:developer Lang:}
		logx.Errorw("DistributePointsMq sendNotificationMessage error", logx.Field("detail", err.Error()))
	}

	return nil
}

func (l *DistributePointsMq) getMessages(
	openId string,
	userAsset *userclient.UpdateUserAssetReq,
) []*subscribe.Message {
	if userAsset.AssetType != nil && *userAsset.AssetType == globalkey.UserAssetTypePoint {
		if userAsset.SourceType != nil {
			var sType string
			if *userAsset.SourceType == globalkey.UserAssetSourceTypeTask {
				sType = "任务得积分"
			} else if *userAsset.SourceType == globalkey.UserAssetSourceTypeReward {
				sType = "消费得积分"
			}
			amount := " "
			if userAsset.ChangeAmount != nil {
				amount = *userAsset.ChangeAmount
			}

			return []*subscribe.Message{
				{
					ToUser:     openId,
					Page:       "pages/scentrobe/package/message/message-page",
					TemplateID: subscribex.ForumUserPointChangeTemplateID,
					Data: map[string]*subscribe.DataItem{
						"thing5":            {Value: sType},                             // 变动原因
						"character_string2": {Value: amount},                            // 变动数量
						"date4":             {Value: modelx.FormatTime(time.Now(), "")}, // 变动时间
						"thing6":            {Value: "积分可以兑换优惠券、试香等"},                   // 温馨提示
					},
				},
			}
		}
	}

	return nil
}
